package com.song.hbasedemo1;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;

public class Hbasedemo01 {
	
	private static Configuration conf = null;
	
	static{
		Configuration HBASE_CONF = new Configuration();
		HBASE_CONF.set("hbase.zookeeper.quorum","localhost");
		
		conf = HBaseConfiguration.create(HBASE_CONF);
	}
	
	
	public static void createTalbe(String tableName , String[] familys)
	{
		try {
			HBaseAdmin admin  = new HBaseAdmin(conf);
			if(admin.tableExists(tableName))
			{
				System.out.println("tables is exists");
			}else{
				
				HTableDescriptor tableDes = new HTableDescriptor(tableName);
				for(int i = 0; i<familys.length ;i++)
				{
					tableDes.addFamily(new HColumnDescriptor(familys[i]));
				}
				admin.createTable(tableDes);
			}
			
			
		} catch (MasterNotRunningException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (ZooKeeperConnectionException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
	}
	
	public static void addData(String tableName ,String rowKey ,String[] column1 ,String[] val1 , String[] column2 , String[] val2)
	{
		Put put = new Put(org.apache.hadoop.hbase.util.Bytes.toBytes(rowKey));
		
		try {
			HTable table = new HTable(conf, Bytes.toBytes(tableName));
			HColumnDescriptor[] columns = table.getTableDescriptor().getColumnFamilies();
			
			
			for(int i = 0; i<columns.length ;i++)
			{
				String familyName = columns[i].getNameAsString();
				if(familyName.equals("song"))
				{
					for(int j = 0;j<column1.length ;j++)
					{
						put.add(Bytes.toBytes(familyName),Bytes.toBytes(column1[j]),Bytes.toBytes(val1[j]));
					}
				}
				if(familyName.equals("yang"))
				{
					for(int j = 0;j<column2.length ;j++)
					{
						put.add(Bytes.toBytes(familyName),Bytes.toBytes(column2[j]),Bytes.toBytes(val2[j]));
					}
				}
			}
			
			table.put(put);
			
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		
		
	}
	
	
	public static void getResult(String tableName , String rowKey)
	{
		
		try {
			Get get = new Get(Bytes.toBytes(rowKey));
			HTable table = new HTable(conf, Bytes.toBytes(tableName));
			
			Result  result = table.get(get);
			
			for(KeyValue kv : result.list())
			{
				System.out.println( "family :" +Bytes.toString(kv.getFamily()));
				System.out.println( "qualifier :" +Bytes.toString(kv.getQualifier()));
				System.out.println(" val is :" + Bytes.toString(kv.getValue()));
				System.out.println(" timestamp :"+kv.getTimestamp());
				System.out.println("-------------------------------------------------");
			}
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		
	}
	
	
	public  static void getAllScan(String tableName)
	{
		Scan scan = new Scan();
		ResultScanner rs = null;
		try {
			HTable table  = new HTable(conf, Bytes.toBytes(tableName));
			rs =table.getScanner(scan);
			
			for(Result r :rs)
			{
				for(KeyValue kv : r.list())
				{
					System.out.println( "family :" +Bytes.toString(kv.getFamily()));
					System.out.println( "qualifier :" +Bytes.toString(kv.getQualifier()));
					System.out.println(" val is :" + Bytes.toString(kv.getValue()));
					System.out.println(" timestamp :"+kv.getTimestamp());
					System.out.println("-------------------------------------------------");
				}
			}
			
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
	}
	
	public static void main(String[] args) {
//		createTalbe("ouyang", new String[]{"yang","ou","song","good"});
		
//		addData("ouyang", "data1", new String[]{"test1","test2"}, new String[]{"aaa","bbb"},  new String[]{"test3","test4"}, new String[]{"ccc","ddd"});
		
//		getResult("ouyang", "data1");
		
		getAllScan("ouyang");
	}
}
